package git.soulbgm.hbase;

import git.soulbgm.hbase.annotation.HBaseColumn;
import git.soulbgm.hbase.annotation.Rowkey;
import git.soulbgm.utils.ClassUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.util.Bytes;

import java.lang.reflect.Field;
import java.util.Date;

/**
 * HBase通过自定义注解对Bean的处理工具类
 *
 * @author SoulBGM
 * @version 1.0
 * @date 2020/6/18 17:39
 */
@Slf4j
public class HBaseBeanUtil {

    /**
     * 获得实体类中rowkey的值
     *
     * @param t 实体类
     * @return rowkey值
     */
    public static <T> String getRowkey(T t) {
        Class<?> clazz = t.getClass();
        Field[] fields = clazz.getDeclaredFields();
        for (Field field : fields) {
            if (ClassUtil.existFieldAnnotation(field, Rowkey.class)) {
                return String.valueOf(ClassUtil.executeGetMethod(t, field));
            }
        }
        return null;
    }

    /**
     * 实体类对象转为存入HBase数据的Put对象
     *
     * @param t 实体类
     * @return Put对象
     */
    public static <T> Put entityToPut(T t) {
        String rowkey = getRowkey(t);
        if (rowkey == null) {
            return null;
        }
        Put put = new Put(Bytes.toBytes(rowkey));
        Class<?> clazz = t.getClass();
        Field[] fields = clazz.getDeclaredFields();
        for (Field field : fields) {
            if (!ClassUtil.existFieldAnnotation(field, HBaseColumn.class)) {
                continue;
            }
            HBaseColumn column = ClassUtil.getFieldAnnotation(field, HBaseColumn.class);
            String family = column.family();
            String qualifier = column.qualifier();
            if ("".equals(qualifier)) {
                qualifier = field.getName();
            }
            Object value = ClassUtil.executeGetMethod(t, field);
            String valueStr = null;
            if (value == null || StringUtils.isBlank(valueStr = value.toString())) {
                continue;
            }
            Class<?> valueClass = value.getClass();
            if (valueClass.isArray()) {
                log.error("不支持{}类型进行转换Put类", valueClass.getTypeName());
            }
            if (valueClass == Date.class) {
                value = ((Date) value).getTime();
                valueStr = value.toString();
            }
            put.addColumn(Bytes.toBytes(family), Bytes.toBytes(qualifier), Bytes.toBytes(valueStr));
        }
        return put;
    }

    /**
     * 将HBase查询的结果转为实体类对象
     *
     * @param result HBase查询的结果
     * @param clazz  需要生成实体类的Class对象
     * @return 转换为实体类的对象
     */
    public static <T> T resultToEntity(Result result, Class<T> clazz) {
        try {
            T obj = clazz.newInstance();
            Field[] fields = clazz.getDeclaredFields();
            for (Field field : fields) {
                Object value = null;
                Class<?> type = field.getType();
                if (ClassUtil.existFieldAnnotation(field, Rowkey.class)) {
                    value = strToObj(new String(result.getRow()), type);
                } else if (ClassUtil.existFieldAnnotation(field, HBaseColumn.class)) {
                    HBaseColumn column = ClassUtil.getFieldAnnotation(field, HBaseColumn.class);
                    String family = column.family();
                    String qualifier = column.qualifier();
                    if ("".equals(qualifier)) {
                        qualifier = field.getName();
                    }
                    byte[] valueByteArray = result.getValue(Bytes.toBytes(family), Bytes.toBytes(qualifier));
                    String hbaseSourceVal = new String(valueByteArray);
                    value = strToObj(hbaseSourceVal, type);
                }
                if (value != null) {
                    ClassUtil.executeSetMethod(obj, field, value);
                }
            }
            return obj;
        } catch (Exception e) {
            e.printStackTrace();
        }
        return null;
    }

    /**
     * 字符串转各种类型
     *
     * @param str   字符串
     * @param clazz 类型的class
     * @return 符合clazz的对象
     */
    public static Object strToObj(String str, Class clazz) {
        if (clazz == null || str == null) {
            return null;
        }
        if (clazz == byte.class || clazz == Byte.class) {
            return Byte.parseByte(str);
        } else if (clazz == short.class || clazz == Short.class) {
            return Short.parseShort(str);
        } else if (clazz == int.class || clazz == Integer.class) {
            return Integer.parseInt(str);
        } else if (clazz == long.class || clazz == Long.class) {
            return Long.parseLong(str);
        } else if (clazz == float.class || clazz == Float.class) {
            return Float.parseFloat(str);
        } else if (clazz == double.class || clazz == Double.class) {
            return Double.parseDouble(str);
        } else if (clazz == boolean.class || clazz == Boolean.class) {
            return Boolean.parseBoolean(str);
        } else if (clazz == char.class || clazz == Character.class) {
            return str.charAt(0);
        } else if (clazz == char[].class || clazz == Character[].class) {
            return str.toCharArray();
        } else if (clazz == Date.class) {
            long timestamp = Long.parseLong(str);
            return new Date(timestamp);
        }
        return str;
    }

}
